package org.uiot.process

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig
import org.uiot.resource.RmqInst

trait RmqInputTraits {

    RmqInst rmqInput
    String sourceQueue

    RMQSource<String> getRmqSource() {
        def rmqConnectionConfig = new RMQConnectionConfig.Builder()
                .setUserName(rmqInput.username)
                .setPassword(rmqInput.password)
                .setVirtualHost("/")//multi-tenant
                .setHost(rmqInput.host)
                .setPort(rmqInput.port)
                .build()

        new RMQSource<String>(
                rmqConnectionConfig,// config for the RabbitMQ connection
                sourceQueue,// name of the RabbitMQ queue to consume
                false,// use correlation ids; can be false if only at-least-once is required
                new SimpleStringSchema())
    }
}